package com.myself.storm_test;

import java.util.HashMap;
import java.util.Map;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
/**
 * 统计每个word出现的次数
 */
@SuppressWarnings("serial")
public class WordCountBolt extends BaseBasicBolt {

	Map<String, Integer> counts = new HashMap<String, Integer>();

	public void execute(Tuple tuple, BasicOutputCollector collector) {
		String word = tuple.getString(0);
		System.out.println(this + "---------------" + word);
		Integer count = counts.get(word);
		if (count == null){
			count = 1;
		}else{
			count++;
		}
		counts.put(word, count);
		collector.emit(new Values(word, count));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word", "count"));
	}
}
